package com.atguigu.flink09;

import com.atguigu.beans.WaterSensor;
import com.atguigu.func.WaterSensorMapFunction;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @author Felix
 * @date 2024/2/29
 * 该案例演示了算子状态的广播状态
 * 需求：水位超过指定的阈值发送告警，阈值可以动态修改
 */
public class Flink02_opeState_BroadcastState {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(1);

        //env.setStateBackend(new HashMapStateBackend());
        //env.setStateBackend(new EmbeddedRocksDBStateBackend());

        //TODO 2.从指定的网络端口读取水位数据
        SingleOutputStreamOperator<WaterSensor> wsDS = env
                .socketTextStream("hadoop102", 8888)
                .map(new WaterSensorMapFunction());
        //TODO 3.从指定的网络端口读取阈值
        DataStreamSource<String> configDS = env
                .socketTextStream("hadoop102", 8889);

        //TODO 4.对阈值流数据进行广播
        //4.1 声明广播状态描述器
        //注意：广播状态中存放数据的类型只能是kv形式，广播流的并行度只能是1

        MapStateDescriptor<String, Integer> mapStateDescriptor
                = new MapStateDescriptor<String, Integer>("mapStateDescriptor",String.class, Integer.class);

        BroadcastStream<String> broadcastDS
                = configDS.broadcast(mapStateDescriptor);

        //TODO 5.将水位流(非广播流)和配置流(广播流)进行关联
        BroadcastConnectedStream<WaterSensor, String> connectDS = wsDS.connect(broadcastDS);

        //TODO 6.对关联后的数据进行处理
        SingleOutputStreamOperator<String> processDS = connectDS.process(
                new BroadcastProcessFunction<WaterSensor, String, String>() {

                    //处理主流数据
                    @Override
                    public void processElement(WaterSensor ws, BroadcastProcessFunction<WaterSensor, String, String>.ReadOnlyContext ctx, Collector<String> out) throws Exception {
                        //获取当前传感器采集到的水位值
                        Integer vc = ws.vc;
                        //获取广播状态
                        ReadOnlyBroadcastState<String, Integer> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
                        //从广播状态中获取阈值
                        Integer threshold = broadcastState.get("threshold") == null ? 0 : broadcastState.get("threshold");
                        if (vc > threshold) {
                            out.collect("当前水位值是" + vc + "，超出了阈值" + threshold);
                        }
                    }

                    //处理广播流数据
                    @Override
                    public void processBroadcastElement(String value, BroadcastProcessFunction<WaterSensor, String, String>.Context ctx, Collector<String> out) throws Exception {
                        //获取广播状态
                        BroadcastState<String, Integer> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
                        //将从配置流中读取到的阈值信息放到广播状态中
                        broadcastState.put("threshold", Integer.valueOf(value));
                    }
                }
        );

        //TODO 7.输出
        processDS.print();

        //TODO 8.提交
        env.execute();

    }
}
